package com.mini.grpc.transport.netty;

import com.mini.grpc.common.Metadata;
import com.mini.grpc.common.MethodDescriptor;
import com.mini.grpc.common.Status;
import com.mini.grpc.transport.ClientStream;
import com.mini.grpc.transport.ClientStreamListener;
import com.mini.grpc.transport.ClientTransport;
import com.mini.grpc.transport.ClientTransport.ConnectivityState;
import com.mini.grpc.transport.TransportListener;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http2.*;
import io.netty.handler.ssl.ApplicationProtocolConfig;
import io.netty.handler.ssl.ApplicationProtocolNames;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.SupportedCipherSuiteFilter;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.SocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.net.ssl.SSLException;

/**
 * NettyClientTransport实现了基于Netty的客户端传输。
 */
public class NettyClientTransport implements ClientTransport {
    
    private static final Logger logger = LoggerFactory.getLogger(NettyClientTransport.class);
    
    private final SocketAddress address;
    private final TransportListener listener;
    private final boolean useSSL;
    
    private EventLoopGroup group;
    private Channel channel;
    private NettyClientHandler handler;
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean shutdown = new AtomicBoolean(false);
    private final CountDownLatch terminationLatch = new CountDownLatch(1);
    private volatile ConnectivityState state = ConnectivityState.IDLE;
    
    /**
     * 创建一个新的NettyClientTransport
     *
     * @param address 服务端地址
     * @param listener 传输监听器
     * @param useSSL 是否使用SSL
     */
    public NettyClientTransport(SocketAddress address, TransportListener listener, boolean useSSL) {
        this.address = address;
        this.listener = listener;
        this.useSSL = useSSL;
    }
    
    @Override
    public <ReqT, RespT> ClientStream<ReqT, RespT> newStream(
            MethodDescriptor<ReqT, RespT> method,
            Metadata headers,
            ClientStreamListener<RespT> streamListener) {
        logger.info("=== NettyClientTransport.newStream called for method: {} ===", method.getFullMethodName());
        
        if (state != ConnectivityState.READY) {
            logger.error("Transport is not ready, current state: {}", state);
            throw new IllegalStateException("Transport is not ready");
        }
        
        logger.info("Creating new stream with handler: {}", handler);
        
        // 创建新的流
        NettyClientStream<ReqT, RespT> stream = new NettyClientStream<>(
                handler.nextStreamId(),
                handler,
                method);
        
        logger.info("Created stream with ID: {}", stream.getStreamId());
        
        // 设置监听器
        stream.setListener(streamListener);
        
        // 将流添加到处理器
        handler.addStream(stream.getStreamId(), stream);
        
        logger.info("About to send headers for stream: {}", stream.getStreamId());
        
        // 开始流
        stream.sendHeaders(headers);
        
        logger.info("Stream created and headers sent for stream: {}", stream.getStreamId());
        
        return stream;
    }
    
    @Override
    public boolean start() {
        if (!started.compareAndSet(false, true)) {
            return false;
        }
        
        try {
            group = new NioEventLoopGroup();
            
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        if (useSSL) {
                            SslContext sslCtx = SslContextBuilder.forClient()
                                .trustManager(InsecureTrustManagerFactory.INSTANCE)
                                .ciphers(Http2SecurityUtil.CIPHERS, SupportedCipherSuiteFilter.INSTANCE)
                                .applicationProtocolConfig(new ApplicationProtocolConfig(
                                    ApplicationProtocolConfig.Protocol.ALPN,
                                    ApplicationProtocolConfig.SelectorFailureBehavior.NO_ADVERTISE,
                                    ApplicationProtocolConfig.SelectedListenerFailureBehavior.ACCEPT,
                                    ApplicationProtocolNames.HTTP_2))
                                .build();
                            ch.pipeline().addLast(sslCtx.newHandler(ch.alloc(), address.toString(), 0));
                        }
                        
                        // 创建HTTP/2连接和编解码器
                        Http2Connection connection = new DefaultHttp2Connection(false);
                        Http2ConnectionEncoder encoder = new DefaultHttp2ConnectionEncoder(connection, new DefaultHttp2FrameWriter());
                        Http2ConnectionDecoder decoder = new DefaultHttp2ConnectionDecoder(connection, encoder, new DefaultHttp2FrameReader());
                        
                        // 创建初始设置
                        Http2Settings initialSettings = new Http2Settings()
                            .initialWindowSize(65535)
                            .maxConcurrentStreams(1000)
                            .maxFrameSize(16384)
                            .maxHeaderListSize(8192);
                        
                        // 配置HTTP/2处理器
                        handler = new NettyClientHandler(decoder, encoder, initialSettings);
                        decoder.frameListener(handler);
                        
                        ch.pipeline().addLast(handler);
                        
                        logger.info("NettyClientHandler created and added to pipeline: {}", handler);
                    }
                });
            
            // 连接到服务器
            setState(ConnectivityState.CONNECTING);
            ChannelFuture f = b.connect(address).sync();
            channel = f.channel();
            
            // 等待一小段时间确保handler被正确初始化
            Thread.sleep(100);
            
            // 连接成功
            setState(ConnectivityState.READY);
            listener.transportReady();
            
            logger.info("Client connected to {}, handler: {}", address, handler);
            
            return true;
        } catch (Exception e) {
            logger.error("Failed to start client", e);
            shutdown(Status.UNAVAILABLE.withDescription("Failed to start client: " + e.getMessage()));
            return false;
        }
    }
    
    @Override
    public void shutdown(Status status) {
        if (!shutdown.compareAndSet(false, true)) {
            return;
        }
        
        logger.info("Shutting down client: {}", status);
        
        try {
            setState(ConnectivityState.SHUTDOWN);
            
            if (channel != null) {
                channel.close().sync();
            }
            
            if (group != null) {
                group.shutdownGracefully(0, 5, TimeUnit.SECONDS);
            }
            
            listener.transportTerminated(status);
        } catch (Exception e) {
            logger.error("Error during client shutdown", e);
        } finally {
            terminationLatch.countDown();
        }
    }
    
    @Override
    public boolean isShutdown() {
        return shutdown.get();
    }
    
    @Override
    public boolean isTerminated() {
        return terminationLatch.getCount() == 0;
    }
    
    @Override
    public void awaitTermination() throws InterruptedException {
        terminationLatch.await();
    }
    
    @Override
    public ConnectivityState getState() {
        return state;
    }
    
    /**
     * 设置连接状态
     *
     * @param newState 新状态
     */
    private void setState(ConnectivityState newState) {
        ConnectivityState oldState = state;
        state = newState;
        
        if (oldState != newState) {
            logger.info("Transport state changed from {} to {}", oldState, newState);
        }
    }
} 